package org.tao.lightningmq.client.consumer;

import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.client.netty.BrokerRemoteRespHandler;
import org.tao.lightningmq.common.dto.*;
import org.tao.lightningmq.common.enums.*;
import org.tao.lightningmq.common.event.EventBus;
import org.tao.lightningmq.common.remote.BrokerNettyRemoteClient;
import org.tao.lightningmq.common.remote.NameServerNettyRemoteClient;
import org.tao.lightningmq.common.tcp.TcpMsg;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @Author lht
 * @date 2025/1/19 - 19:53
 * @description:
 */
public class DefaultMqConsumer {

    private static final Logger LOGGER = LoggerFactory.getLogger(DefaultMqConsumer.class);
    private final static int EACH_BATCH_PULL_MSG_INTER = 100; //如果broker有数据，每间隔100ms拉一批
    private final static int EACH_BATCH_PULL_MSG_INTER_WHEN_NO_MSG = 1000; //如果broker无数据，每间隔1s拉一批

    private String nameServerIp;
    private int nameServerPort;
    private String nameServerUser;
    private String nameServerPassword;
    private String topic;
    private String consumeGroup;
    private String brokerRole = "single";
    private Integer queueId;
    private Integer batchSize;
    private String brokerClusterGroup;
    private NameServerNettyRemoteClient nameServerNettyRemoteClient;
    private List<String> brokerAddressList;
    private List<String> masterAddressList;
    private List<String> slaveAddressList;
    private MessageConsumeListener messageConsumeListener;
    private Map<String, BrokerNettyRemoteClient> brokerNettyRemoteClientMap = new ConcurrentHashMap<>();
    private final CountDownLatch countDownLatch = new CountDownLatch(1);
    private List<String> writableBrokerNodes = Collections.emptyList();
    private List<String> readableBrokerNodes = Collections.emptyList();
    private final Random random = new Random();
    private String nameserverAddrStr;
    private final Map<String, NameServerNettyRemoteClient> nameServerNettyRemoteClientMap = new ConcurrentHashMap<>();

    public void start() throws InterruptedException {
//        nameServerNettyRemoteClient = new NameServerNettyRemoteClient(nameServerIp, nameServerPort);
//        nameServerNettyRemoteClient.buildConnection();
//        boolean isRegistrySuccess = this.doRegistry();
        boolean isRegistrySuccess = false;
        for (String nameserverAddr : nameserverAddrStr.split(";")) {
            String[] split = nameserverAddr.split(":");
            NameServerNettyRemoteClient ns = new NameServerNettyRemoteClient(split[0], Integer.parseInt(split[1]));
            ns.buildConnection();
            ns.setNameServerLogoutCallBack(() -> {
                ns.buildConnection();
                doRegistry(ns);
            });
            isRegistrySuccess |= doRegistry(ns);
            LOGGER.info("producer registry to {} result is: {}", nameserverAddr, isRegistrySuccess);
            nameServerNettyRemoteClientMap.put(nameserverAddr, ns);
        }
        if (isRegistrySuccess) {
            this.startHeartBeatTask();
            this.fetchBrokerAddress();
            this.connectBroker();
            // 创建普通topic和重试topic
            this.createTopic(topic);
            // 连接成功后注册消费者组到Broker中, 这里就直接向所有节点注册了，不走主从复制
            this.createConsumeGroupOffsetTable(this.readableBrokerNodes);
            this.startConsumeMsgTask(topic);
            this.startConsumeMsgTask("retry%" + this.getConsumeGroup());
            this.startRefreshBrokerAddressJob();
            countDownLatch.await();
        }
    }

    private void createTopic(String topic) {
        CreateTopicClient createTopicClient = null;
        if (this.writableBrokerNodes != null && !this.writableBrokerNodes.isEmpty()) {
            createTopicClient = new CreateTopicClient(this.writableBrokerNodes.get(0));
        }
        if (createTopicClient != null) {
            // 普通主题
            createTopicClient.createTopic(topic);
            // 重试主题
            createTopicClient.createTopic("retry%" + this.getConsumeGroup());
            // 死信队列
            createTopicClient.createTopic("dead_queue");
            createTopicClient.close();
        }
    }

    private void connectBroker() {
        if ("single".equals(this.getBrokerRole())) {
            if (this.getBrokerAddressList() == null || this.getBrokerAddressList().isEmpty()) {
                throw new RuntimeException("single broker地址不能为空");
            }
            writableBrokerNodes = this.getBrokerAddressList();
            readableBrokerNodes = this.getBrokerAddressList();
        } else {
            if (this.getMasterAddressList() == null || this.getMasterAddressList().isEmpty()) {
                throw new RuntimeException("master broker地址不能为空");
            }
            writableBrokerNodes = this.getMasterAddressList();
            readableBrokerNodes = Stream.of(this.getMasterAddressList(), this.getSlaveAddressList())
                    .flatMap(Collection::stream).collect(Collectors.toList());
        }
        // readableBrokerNodes包含了writableBrokerNodes的节点
        doCheckAndConnect(readableBrokerNodes);
    }

    private void doCheckAndConnect(List<String> brokerAddressList) {
        brokerAddressList.forEach(brokerAddr -> {
            BrokerNettyRemoteClient nettyRemoteClient = this.brokerNettyRemoteClientMap.get(brokerAddr);
            if (nettyRemoteClient != null) {
                if (!nettyRemoteClient.isChannelActive()) {
                    // 不活跃的节点进行剔除
                    nettyRemoteClient.close();
                    this.brokerNettyRemoteClientMap.remove(brokerAddr);
                } else return;
            }
            // 对于未连接的broker，建立连接
            String[] split = brokerAddr.split(":");
            String ip = split[0], port = split[1];
            BrokerNettyRemoteClient brokerNettyRemoteClient = new BrokerNettyRemoteClient(ip, Integer.parseInt(port));
            brokerNettyRemoteClient.buildConnection(new BrokerRemoteRespHandler(new EventBus("consumer-client-eventbus")));
            LOGGER.info("connect broker:{} success", brokerAddr);
            this.brokerNettyRemoteClientMap.put(brokerAddr, brokerNettyRemoteClient);
        });
    }

    private void createConsumeGroupOffsetTable(List<String> brokerAddrs) {
        RegistryConsumerGroupReqDTO reqDTO = new RegistryConsumerGroupReqDTO();
        reqDTO.setTopic(this.topic);
        reqDTO.setConsumeGroup(this.consumeGroup);
        reqDTO.setBatchSize(this.batchSize);
        for (String brokerAddr : brokerAddrs) {
            BrokerNettyRemoteClient brokerNettyRemoteClient = this.brokerNettyRemoteClientMap.get(brokerAddr);
            reqDTO.setMsgId(UUID.randomUUID().toString());
            TcpMsg tcpMsg = new TcpMsg(BrokerEventCode.REGISTRY_CONSUMER_GROUP.getCode(), JSON.toJSONBytes(reqDTO));
            TcpMsg resp = brokerNettyRemoteClient.sendSyncMsg(tcpMsg, reqDTO.getMsgId());
            if (resp != null) {
                RegistryConsumerGroupRespDTO respDTO = JSON.parseObject(resp.getBody(), RegistryConsumerGroupRespDTO.class);
                LOGGER.info("注册到 {} 的consumerGroup offset table的结果: {}", brokerAddr, Boolean.TRUE.equals(respDTO.getSuccess()));
            }
        }
    }

    private void startRefreshBrokerAddressJob() {
        new Thread(() -> {
            while (true) {
                try {
                    TimeUnit.SECONDS.sleep(30);
                    this.fetchBrokerAddress();
                    this.connectBroker();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }, "consumer-refresh-broker-address-job").start();
    }

    private void startConsumeMsgTask(String topic) {
        new Thread(() -> {
            while (true) {
                try {
                    List<String> writableNodes = this.writableBrokerNodes;
                    List<String> readableNodes = this.readableBrokerNodes;
                    if (writableNodes == null || writableNodes.isEmpty()) {
                        TimeUnit.MILLISECONDS.sleep(EACH_BATCH_PULL_MSG_INTER_WHEN_NO_MSG);
                        LOGGER.warn("broker address is empty!");
                        continue;
                    }

                    BrokerNettyRemoteClient writableNodeClient = this.brokerNettyRemoteClientMap.get(writableNodes.get(0));
                    BrokerNettyRemoteClient readableNodeClient = getListElementRandom(readableNodes);

                    ConsumeMsgReqDTO consumeMsgReqDTO = new ConsumeMsgReqDTO();
                    consumeMsgReqDTO.setConsumeGroup(consumeGroup);
                    consumeMsgReqDTO.setBatchSize(batchSize);
                    consumeMsgReqDTO.setTopic(topic);
                    String msgId = UUID.randomUUID().toString();
                    consumeMsgReqDTO.setMsgId(msgId);

                    TcpMsg pullReqMsg = new TcpMsg(BrokerEventCode.CONSUME_MSG.getCode(), JSON.toJSONBytes(consumeMsgReqDTO));
                    TcpMsg pullMsgResp = readableNodeClient.sendSyncMsg(pullReqMsg, msgId);

                    List<ConsumeMsgRespDTO> consumeMsgRespDTOS = null;
                    // FIXME nullpointer
                    ConsumeMsgBaseRespDTO consumeMsgBaseRespDTO = JSON.parseObject(pullMsgResp.getBody(), ConsumeMsgBaseRespDTO.class);
                    if (consumeMsgBaseRespDTO != null) {
                        consumeMsgRespDTOS = consumeMsgBaseRespDTO.getConsumeMsgRespDTOList();
                    }
                    boolean brokerHasData = false;
                    if (consumeMsgRespDTOS != null && !consumeMsgRespDTOS.isEmpty()) {
                        for (ConsumeMsgRespDTO consumeMsgRespDTO : consumeMsgRespDTOS) {
                            List<ConsumeMsgCommitLogDTO> commitLogContentList = consumeMsgRespDTO.getCommitLogContentList();
                            if (commitLogContentList == null || commitLogContentList.isEmpty()) {
                                continue;
                            }
                            List<ConsumeMessage> consumeMessages = new ArrayList<>();
                            for (ConsumeMsgCommitLogDTO consumeMsgCommitLogDTO : commitLogContentList) {
                                ConsumeMessage consumeMessage = new ConsumeMessage();
                                consumeMessage.setConsumeMsgCommitLogDTO(consumeMsgCommitLogDTO);
                                consumeMessages.add(consumeMessage);
                            }
                            brokerHasData = true;
                            ConsumeResult consumeResult = messageConsumeListener.consume(consumeMessages);
                            if (consumeResult.getStatus() == ConsumeResult.ConsumeResultStatus.CONSUME_SUCCESS) {
                                //消费成功，提交消费进度
                                this.setQueueId(consumeMsgRespDTO.getQueueId());
                                String ackMsgId = UUID.randomUUID().toString();
                                ConsumeMsgAckReqDTO consumeMsgAckReqDTO = new ConsumeMsgAckReqDTO();
                                consumeMsgAckReqDTO.setAckCount(consumeMessages.size());
                                consumeMsgAckReqDTO.setConsumeGroup(this.getConsumeGroup());
                                consumeMsgAckReqDTO.setTopic(topic);
                                consumeMsgAckReqDTO.setQueueId(consumeMsgRespDTO.getQueueId());
                                consumeMsgAckReqDTO.setMsgId(ackMsgId);
                                consumeMsgAckReqDTO.setStartMsgConsumeQueueOffset(commitLogContentList.get(0).getMsgConsumeQueueOffset());
                                TcpMsg ackReq = new TcpMsg(BrokerEventCode.CONSUME_SUCCESS_MSG.getCode(), JSON.toJSONBytes(consumeMsgAckReqDTO));
                                TcpMsg ackResponse = writableNodeClient.sendSyncMsg(ackReq, ackMsgId);
                                ConsumeMsgAckRespDTO consumeMsgAckRespDTO = JSON.parseObject(ackResponse.getBody(), ConsumeMsgAckRespDTO.class);
                                if (AckStatus.SUCCESS.getCode() == consumeMsgAckRespDTO.getAckStatus()) {
                                    LOGGER.info("consume ack success!");
                                } else {
                                    LOGGER.info("consume ack fail!");
                                }
                            } else {
                                //消费失败，回应ACK然后放到重试队列中
                                this.queueId = consumeMsgRespDTO.getQueueId();
                                ConsumeMsgRetryReqDTO consumeMsgRetryReqDTO = new ConsumeMsgRetryReqDTO();
                                List<ConsumeMsgRetryReqDetailDTO> consumeMsgRetryReqDTOS = commitLogContentList
                                        .stream()
                                        .map(consumeMsgCommitLogDTO -> {
                                            ConsumeMsgRetryReqDetailDTO consumeMsgRetryReqDetailDTO = new ConsumeMsgRetryReqDetailDTO();
                                            consumeMsgRetryReqDetailDTO.setConsumerGroup(this.getConsumeGroup());
                                            consumeMsgRetryReqDetailDTO.setTopic(topic);
                                            consumeMsgRetryReqDetailDTO.setQueueId(consumeMsgRespDTO.getQueueId());
                                            consumeMsgRetryReqDetailDTO.setCommitLogOffset(consumeMsgCommitLogDTO.getCommitLogOffset());
                                            consumeMsgRetryReqDetailDTO.setCommitLogMsgLength(consumeMsgCommitLogDTO.getCommitLogSize());
                                            consumeMsgRetryReqDetailDTO.setRetryTime(consumeMsgCommitLogDTO.getRetryTimes() + 1);
                                            consumeMsgRetryReqDetailDTO.setCommitLogName(consumeMsgCommitLogDTO.getFileName());
                                            return consumeMsgRetryReqDetailDTO;
                                        }).collect(Collectors.toList());
                                consumeMsgRetryReqDTO.setConsumeMsgRetryReqDetailDTOList(consumeMsgRetryReqDTOS);
                                consumeMsgRetryReqDTO.setMsgId(UUID.randomUUID().toString());
                                TcpMsg tcpMsg = new TcpMsg(BrokerEventCode.CONSUME_LATER_MSG.getCode(), JSON.toJSONBytes(consumeMsgRetryReqDTO));
                                TcpMsg resp = writableNodeClient.sendSyncMsg(tcpMsg, consumeMsgRetryReqDTO.getMsgId());
                                LOGGER.info("consume fail, retry later, response:{}", resp);
                            }
                        }
                    }
                    if (brokerHasData) {
                        TimeUnit.MILLISECONDS.sleep(EACH_BATCH_PULL_MSG_INTER);
                    } else {
                        TimeUnit.MILLISECONDS.sleep(EACH_BATCH_PULL_MSG_INTER_WHEN_NO_MSG);
                    }
                } catch (InterruptedException e) {
                    LOGGER.warn("consume error: {}", e);
                    try {
                        TimeUnit.MILLISECONDS.sleep(EACH_BATCH_PULL_MSG_INTER_WHEN_NO_MSG);
                    } catch (InterruptedException ex) {
                        throw new RuntimeException(ex);
                    }
                }
            }
        }, "consume-msg-task").start();
    }

    private BrokerNettyRemoteClient getListElementRandom(List<String> readableNodes) {
        int randomIdx = random.nextInt(readableNodes.size());
        String brokerAddr = readableNodes.get(randomIdx);
        // 随机选取一个活跃的客户端，有的话就返回
        BrokerNettyRemoteClient brokerNettyRemoteClient = this.brokerNettyRemoteClientMap.get(brokerAddr);
        if (brokerNettyRemoteClient != null && brokerNettyRemoteClient.isChannelActive()) {
            return brokerNettyRemoteClient;
        }
        // 随机挑选的客户端是不活跃的，遍历列表挑选
        for (String readableNode : readableNodes) {
            brokerNettyRemoteClient = this.brokerNettyRemoteClientMap.get(brokerAddr);
            if (brokerNettyRemoteClient != null && brokerNettyRemoteClient.isChannelActive()) {
                return brokerNettyRemoteClient;
            }
        }
        throw new RuntimeException("没有可用的读取Broker");
    }

    private void fetchBrokerAddress() {
        NameServerNettyRemoteClient nsClient = getUsefulClient();
        String msgId = UUID.randomUUID().toString();
        PullBrokerIpDTO pullBrokerIpDTO = new PullBrokerIpDTO();
        if (this.brokerClusterGroup != null) {
            this.brokerRole = "master";
            pullBrokerIpDTO.setBrokerClusterGroup(this.brokerClusterGroup);
        }
        pullBrokerIpDTO.setRole(this.brokerRole);
        pullBrokerIpDTO.setMsgId(msgId);
        TcpMsg resp = nsClient.sendSyncMsg(new TcpMsg(NameServerEventCode.PULL_BROKER_IP_LIST.getCode(),
                JSON.toJSONBytes(pullBrokerIpDTO)), msgId);
        //获取broker节点ip地址，并且缓存起来，可能由多个master-broker角色
        PullBrokerIpRespDTO pullBrokerIpRespDTO = JSON.parseObject(resp.getBody(), PullBrokerIpRespDTO.class);
        this.setBrokerAddressList(pullBrokerIpRespDTO.getAddressList());
        this.setMasterAddressList(pullBrokerIpRespDTO.getMasterAddressList());
        this.setSlaveAddressList(pullBrokerIpRespDTO.getSlaveAddressList());
        LOGGER.info("fetch broker address:{},master:{},slave:{}", this.getBrokerAddressList(), this.getMasterAddressList(), this.getSlaveAddressList());
    }

    private NameServerNettyRemoteClient getUsefulClient() {
        for (NameServerNettyRemoteClient value : nameServerNettyRemoteClientMap.values()) {
            if (value.isActive()) {
                return value;
            }
        }
        throw new RuntimeException("没有已连接的Nameserver节点");
    }

    private void startHeartBeatTask() {
        Thread heartBeatTask = new Thread(() -> {
            HeartBeatDTO heartBeatDTO = new HeartBeatDTO();
            while (true) {
                try {
                    TimeUnit.SECONDS.sleep(3);
                    String heartBeatMsgId = UUID.randomUUID().toString();
                    heartBeatDTO.setMsgId(heartBeatMsgId);
//                    TcpMsg heartBeatResponse = nameServerNettyRemoteClient.sendSyncMsg(new TcpMsg(NameServerEventCode.HEART_BEAT.getCode(),
//                            JSON.toJSONBytes(heartBeatDTO)), heartBeatMsgId);
//                    LOGGER.info("heart beat response data is :{}", JSON.parseObject(heartBeatResponse.getBody()));
                    TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.HEART_BEAT.getCode(), JSON.toJSONBytes(heartBeatDTO));
                    for (NameServerNettyRemoteClient nsClient : nameServerNettyRemoteClientMap.values()) {
                        if (nsClient.isActive()) {
                            nsClient.sendAsyncMsg(tcpMsg);
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "consumer-heart-beat-task");
        heartBeatTask.start();
    }

    /**
     * 开启注册
     *
     * @return
     */
    private boolean doRegistry(NameServerNettyRemoteClient client) {
        String registryMsgId = UUID.randomUUID().toString();
        ServiceRegistryReqDTO serviceRegistryReqDTO = new ServiceRegistryReqDTO();
        serviceRegistryReqDTO.setMsgId(registryMsgId);
        serviceRegistryReqDTO.setUser(nameServerUser);
        serviceRegistryReqDTO.setPassword(nameServerPassword);
        serviceRegistryReqDTO.setRegistryType(RegistryTypeEnum.CONSUMER.getCode());
        TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.REGISTRY.getCode(), JSON.toJSONBytes(serviceRegistryReqDTO));
        TcpMsg registryResponse = client.sendSyncMsg(tcpMsg, registryMsgId);
        if (NameServerResponseCode.REGISTRY_SUCCESS.getCode() == registryResponse.getCode()) {
            return true;
        }
        LOGGER.warn("注册账号失败");
        return false;
    }

    public String getNameServerIp() {
        return nameServerIp;
    }

    public void setNameServerIp(String nameServerIp) {
        this.nameServerIp = nameServerIp;
    }

    public int getNameServerPort() {
        return nameServerPort;
    }

    public void setNameServerPort(int nameServerPort) {
        this.nameServerPort = nameServerPort;
    }

    public String getNameServerUser() {
        return nameServerUser;
    }

    public void setNameServerUser(String nameServerUser) {
        this.nameServerUser = nameServerUser;
    }

    public String getNameServerPassword() {
        return nameServerPassword;
    }

    public void setNameServerPassword(String nameServerPassword) {
        this.nameServerPassword = nameServerPassword;
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public String getConsumeGroup() {
        return consumeGroup;
    }

    public void setConsumeGroup(String consumeGroup) {
        this.consumeGroup = consumeGroup;
    }

    public String getBrokerRole() {
        return brokerRole;
    }

    public void setBrokerRole(String brokerRole) {
        this.brokerRole = brokerRole;
    }

    public Integer getQueueId() {
        return queueId;
    }

    public void setQueueId(Integer queueId) {
        this.queueId = queueId;
    }

    public Integer getBatchSize() {
        return batchSize;
    }

    public void setBatchSize(Integer batchSize) {
        this.batchSize = batchSize;
    }

    public String getBrokerClusterGroup() {
        return brokerClusterGroup;
    }

    public void setBrokerClusterGroup(String brokerClusterGroup) {
        this.brokerClusterGroup = brokerClusterGroup;
    }

    public NameServerNettyRemoteClient getNameServerNettyRemoteClient() {
        return nameServerNettyRemoteClient;
    }

    public void setNameServerNettyRemoteClient(NameServerNettyRemoteClient nameServerNettyRemoteClient) {
        this.nameServerNettyRemoteClient = nameServerNettyRemoteClient;
    }

    public List<String> getBrokerAddressList() {
        return brokerAddressList;
    }

    public void setBrokerAddressList(List<String> brokerAddressList) {
        this.brokerAddressList = brokerAddressList;
    }

    public List<String> getMasterAddressList() {
        return masterAddressList;
    }

    public void setMasterAddressList(List<String> masterAddressList) {
        this.masterAddressList = masterAddressList;
    }

    public List<String> getSlaveAddressList() {
        return slaveAddressList;
    }

    public void setSlaveAddressList(List<String> slaveAddressList) {
        this.slaveAddressList = slaveAddressList;
    }

    public MessageConsumeListener getMessageConsumeListener() {
        return messageConsumeListener;
    }

    public void setMessageConsumeListener(MessageConsumeListener messageConsumeListener) {
        this.messageConsumeListener = messageConsumeListener;
    }

    public Map<String, BrokerNettyRemoteClient> getBrokerNettyRemoteClientMap() {
        return brokerNettyRemoteClientMap;
    }

    public void setBrokerNettyRemoteClientMap(Map<String, BrokerNettyRemoteClient> brokerNettyRemoteClientMap) {
        this.brokerNettyRemoteClientMap = brokerNettyRemoteClientMap;
    }

    public String getNameserverAddrStr() {
        return nameserverAddrStr;
    }

    public void setNameserverAddrStr(String nameserverAddrStr) {
        this.nameserverAddrStr = nameserverAddrStr;
    }
}
